package com.zshield.proc;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.zshield.config.KafkaConfig;
import com.zshield.entry.*;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.log4j.Logger;

import java.net.URLEncoder;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class GenericQueryProc<T extends EntryInterface> extends AbstractProcessor<String,String> {
    private ProcessorContext context;
    private Gson gson;
    private JsonParser jp;
    private Class<T> type;
    private String day2;
    private String day1;
    private String day3;
    private boolean offline;
    private KeyValueStore<String,String> kvDay1;
    private KeyValueStore<String,String> kvDay2;
    private KeyValueStore<String,String> kvDay3;
    private KeyValueStore<String,String> kvCurrentDay;
    private Map<String,String> map;
    private static final Logger logger = Logger.getLogger(GenericQueryProc.class);

    GenericQueryProc() {
        gson = new Gson();
    }

    public GenericQueryProc(Class<T> t) {
        this.type = t;
    }

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        kvDay1 = (KeyValueStore<String,String>)context.getStateStore("kvDay1");
        kvDay2 = (KeyValueStore<String,String>)context.getStateStore("qstore");
        kvDay3 = (KeyValueStore<String,String>)context.getStateStore("kvDay3");
        day1 = kvDay1.get("day");
        day2 = kvDay2.get("day");
        day3 = kvDay3.get("day");
        gson = new Gson();
        jp = new JsonParser();
        offline = false;
        map = new ConcurrentHashMap<>();
        context.schedule(60 * 1000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
            for (Map.Entry<String,String> toUpdate : map.entrySet()) {
                context.forward(toUpdate.getKey(), toUpdate.getValue(),"Generic_query_proc");
            }
            map.clear();
            this.context.commit();
        });
    }

    @Override
    public void process(String key, String newValue) {
        try {
            EntryInterface<?> newEntry = gson.fromJson(newValue, type);
            String docId = newEntry.getDocId();
            String processDay = newEntry.getDayTime();
            offline = false;
            if ((day1 == null || processDay.compareTo(day1) > 0) && (day2 == null || processDay.compareTo(day2) > 0) && (day3 == null || processDay.compareTo(day3) > 0)) {
                String maxDay = day1 != null ? day1 : "";
                maxDay = day2 != null && day2.compareTo(maxDay) > 0 ? day2 : maxDay;
                maxDay = day3 != null && day3.compareTo(maxDay) > 0 ? day3 : maxDay;

                if (maxDay.equals(day2)) {
                    day3 = processDay;
                    newDayCome(processDay ,kvDay1, day1, kvDay3);
                } else if (maxDay.equals(day3)) {
                    day1 = processDay;
                    newDayCome(processDay, kvDay2, day2, kvDay1);
                } else if (maxDay.equals(day1) || maxDay.equals("")) {
                    day2 = processDay;
                    newDayCome(processDay, kvDay3, day3, kvDay2);
                }

            } else {
                if (processDay.equals(day1)) {
                    kvCurrentDay = kvDay1;
                } else if (processDay.equals(day2)) {
                    kvCurrentDay = kvDay2;
                } else if (processDay.equals(day3)) {
                    kvCurrentDay = kvDay3;

                }else {
                    offline = true;
                    JsonObject obj = jp.parse(newValue).getAsJsonObject();
                    String ptype = obj.get("PTYPE").getAsString();
                    switch(ptype){
                        case "NetDeny" : context.forward(docId, newValue, "offline_net_deny_query_proc"); break;
                        case "NetIO" : context.forward(docId, newValue, "offline_netio_query_proc"); break;
                        case "FileInOut" : context.forward(docId, newValue, "offline_filesystem_inout_query_proc"); break;
                        case "FilesysTree" : context.forward(docId, newValue, "offline_filesystem_tree_query_proc"); break;
                        case "Filesys" : context.forward(docId, newValue, "offline_filesystem_query_proc"); break;
                        case "FileParse" : context.forward(docId, newValue, "offline_content_analysis_query_proc"); break;
                        case "LnxApp" : context.forward(docId, newValue, "offline_linuxNetIo_app_query_proc"); break;
                        default : logger.error("Unknown PTYPE");
                    }
                }
            }
            if (!offline){
                newValue = updateNewValue(type, kvCurrentDay, newValue, docId, newEntry);
                kvCurrentDay.put(docId,newValue);
                docId = URLEncoder.encode(docId,"UTF-8");
                map.put(docId, newValue);
            }

        }catch (Throwable e){
            logger.error(newValue);
            logger.error("[GenericQueryProc process exception] [The reason for error {" + e + "}");
            KafkaConfig.printErrorLog(logger,e);
        }
    }

    private void newDayCome(String processDay, KeyValueStore<String,String> clearKV, String clearDay, KeyValueStore<String,String> kvCurrentDay) {
        ClearKV(clearKV, clearDay);
        this.kvCurrentDay = kvCurrentDay;
        this.kvCurrentDay.put("day", clearDay);
        logger.info("new date is :" + processDay + ",put (day, " + processDay + ") into " + kvCurrentDay.name());
    }

    String updateNewValue(Class<? extends EntryInterface> type, KeyValueStore<String, String> kv, String newValue, String docId, EntryInterface newEntry) {
        String value = newValue;
        String oldLogValue = kv.get(docId);
        if (oldLogValue != null) {
            EntryInterface entryInterface = this.gson.fromJson(oldLogValue, type);
            entryInterface.update(newEntry);
            value = this.gson.toJson(entryInterface);
        }
        return value;
    }

    private void ClearKV(KeyValueStore<String, String> kv, String clearDay){
        if (clearDay == null){
            return;
        }
        int i = 0;
        long t;
        t = System.currentTimeMillis();
        KeyValueIterator<String, String> it = kv.all();
        while (it.hasNext()) {
            i++;
            kv.delete(it.next().key);
        }
        it.close();
        if(i > 0){
            logger.info(kv.name() + " delete " + i + " pieces,spend " + (System.currentTimeMillis() - t) + "ms,the day is " + clearDay);
        }
    }
}
